package com.example.demo.mqserver.core;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: 张
 * Date: 2024-08-20
 * Time: 20:13
 */

import com.example.demo.common.Consumer;
import com.example.demo.common.ConsumerEnv;
import com.example.demo.common.MqException;
import com.example.demo.mqserver.VirtualHost;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 通过这个类来实现消费消息的核心
 */
public class ConsumerManager {
    // 持有一个上层的应用,用来操作数据
    private VirtualHost parent;
    // 指定一个线程池,负责去执行每个消费者的回调任务
    private ExecutorService workPool = new ForkJoinPool(4);
    // 存放令牌的阻塞队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    // 扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost parent) {
        this.parent = parent;
        scannerThread = new Thread(() -> {
            while(true) {
                try{
                    //1.拿到令牌
                    String queueName = tokenQueue.take();
                    //2.根据令牌找到对应队列
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[ConsumerManager] 去令牌后发现,该队列名不存在!! queueName = " + queueName);
                    }
                    //3.从队列中消费一个消息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        // 把线程设置为后台线程(后台线程不会影响到整个进程的结束)
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    //往阻塞队列中插入令牌(交换机发送消息,队列接收到消息的时候)
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    //添加消费者(消费者订阅消息)
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        //找到对应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if(queue == null) {
            throw new MqException("[ConsumerManager] 队列不存在!! queueName = " + queueName);
        }
        //构造ConsumerEnv(消费者)
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);
            // 如果当前队列中已经有了一些消息了,就需要立即消费掉 --- 生产者先上线玩了一会,消费者后上线直接猛吃
            int count = parent.getMemoryDataCenter().getMessageCount(queueName);
            //直接全吃
            for (int i = 0; i < count; i++) {
                consumeMessage(queue);//消费消息
            }
        }
    }

    //调用一次就消费一条消息(取出一条消息给线程池)
    private void consumeMessage(MSGQueue queue) {
        // 1.按照轮询的方式,找出个消费者来  luckyDog这个真是幸运儿
        ConsumerEnv luckyDog = queue.choseConsumer();
        if (luckyDog == null) {
            //当前队列没有消费者订阅,暂时不进行消费(生产者先上线)
            return;
        }
        // 2.从队列中取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message == null) {
            //当前队列中没有消息(消费者先上线)
            return;
        }
        // 3.把消息带入到消费者的回调方法中,丢给线程池执行
        workPool.submit(() -> {
           try{
               //这块得留个心眼盯着
               //1.将消息放到待确认的集合中 -- 执行回调之前
               parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(),message);
               //2.执行回调 -- 消费消息的核心
               luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(),message.getBasicProperties(),
                       message.getBody());
               //3.判断ConsumerEnv中的autoAck,如果是自动应答就删除消息
               //  如果是手动应答,当前就不进行处理,交给后续消费者调用basicAck方法来进行处理
               if(luckyDog.isAutoAck()) {
                   //1,删除硬盘上的消息
                   if (message.getDeliverMode() == 2) {
                       parent.getDiskDataCenter().deleteMessage(queue,message);
                   }
                   //2,删除待确认消息中的消息
                   parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(),message.getMessageId());
                   //3,删除消息中心中的消息
                   parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                   System.out.println("[ConsumerManager] 消息被成功消费!! queueName = " + queue.getName());
               }
           } catch (Exception e) {
               e.printStackTrace();
           }
        });
    }


}
